/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.asp.bigdata.flink.kafka2hdfs.task;

import com.alibaba.fastjson.JSON;
import com.asp.bigdata.flink.kafka2hdfs.gz.GzipBulkStringWriterFactory;
import com.asp.bigdata.flink.kafka2hdfs.pojo.Event;
import com.asp.bigdata.flink.kafka2hdfs.utils.JsonBucketAssigner;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.io.FileInputStream;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Map;
import java.util.Properties;

@Deprecated
public class kafka2hdfsJson2csv {

	public static void main(String[] args) throws Exception {

		Properties jobConfig = new Properties();
		jobConfig.load(new FileInputStream(args[0]));

		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.enableCheckpointing(Long.valueOf(jobConfig.getProperty("checkpoint.interval")));
		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.valueOf(jobConfig.getProperty("fixed-delay.attempts")), Integer.valueOf(jobConfig.getProperty("fixed-delay.delay"))));

		CheckpointConfig checkpointConfig = env.getCheckpointConfig();
		checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
		// 任务取消时清除外部保存的检查点(非暴力停止如yarn kill)
		checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
		checkpointConfig.setTolerableCheckpointFailureNumber(10);

		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", jobConfig.getProperty("bootstrap.servers"));
		properties.setProperty("group.id", jobConfig.getProperty("group.id"));
		if(jobConfig.containsKey("security.protocol")) {
			properties.setProperty("security.protocol",jobConfig.getProperty("security.protocol"));
		}
		if(jobConfig.containsKey("sasl.mechanism")) {
		    properties.setProperty("sasl.mechanism", jobConfig.getProperty("sasl.mechanism"));
		}
		// properties.setProperty("sasl.kerberos.service.name", "kafka");

		FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(jobConfig.getProperty("topic"), new SimpleStringSchema(), properties);
		consumer.setStartFromGroupOffsets();

		DataStreamSource<String> streamSource = env.addSource(consumer, "Kafka");

        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(jobConfig.getProperty("time.format"));
		// 配置watermark，指定事件时间列
		SingleOutputStreamOperator<Event> operatorWithWM = streamSource.flatMap(
				(FlatMapFunction<String, Event>) (value, out) -> {
					Map valueMap = (Map)JSON.parse(value);
					long time;
					try {
						String even_time = (String)valueMap.get(jobConfig.getProperty("time.field"));
						time =  simpleDateFormat.parse(even_time).getTime();
					} catch (Exception e) {
						e.printStackTrace();
						time = Timestamp.valueOf("1970-01-01 00:00:00").getTime();
					}
					out.collect(new Event(StringUtils.join(valueMap.values(),"|"),time,null));
				})
				.returns(TypeInformation.of(Event.class))
				.assignTimestampsAndWatermarks(WatermarkStrategy
						.<Event>forBoundedOutOfOrderness(Duration.ofMinutes(5))
						.withTimestampAssigner((event, timestamp) -> {
							return event.getTime();
						})  // 指定事件时间列
						.withIdleness(Duration.ofMinutes(1))  // 1分钟内仍未收到数据，标记为空闲数据源
				);

		// 转换成String
//		SingleOutputStreamOperator<String> operatorString = operatorWithWM.map(
//				(MapFunction<Event, String>) value -> value.toString()
//		);
		SingleOutputStreamOperator<Event> operatorString = operatorWithWM;

		// 添加gz后缀
		OutputFileConfig config = OutputFileConfig.builder().withPartSuffix(".gz").build();

		final StreamingFileSink<Event> hdfsSink = StreamingFileSink
				.forBulkFormat(
						new Path(jobConfig.getProperty("path")),
						new GzipBulkStringWriterFactory<Event>())
				.withBucketAssigner(new JsonBucketAssigner(jobConfig))
				.withRollingPolicy(OnCheckpointRollingPolicy.build())
				.withOutputFileConfig(config)
				.build();

		operatorString.addSink(hdfsSink).name("hdfs");

		// execute program
		env.execute(jobConfig.getProperty("app.name"));
	}
}
